package com.wejoy.video.middle.station.data.app;

import com.wejoy.video.middle.station.data.event.wjstream.WjDefaultStreamMessageListenerContainer;
import com.wejoy.video.middle.station.data.constant.RedisKeys;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.autoconfigure.data.redis.RedisProperties;
import org.springframework.boot.context.event.ApplicationStartedEvent;
import org.springframework.context.ApplicationEvent;
import org.springframework.context.ApplicationListener;
import org.springframework.context.Lifecycle;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.context.event.ContextClosedEvent;
import org.springframework.data.redis.cache.RedisCacheConfiguration;
import org.springframework.data.redis.cache.RedisCacheManager;
import org.springframework.data.redis.cache.RedisCacheWriter;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.RedisStandaloneConfiguration;
import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
import org.springframework.data.redis.connection.lettuce.LettucePoolingClientConfiguration;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.GenericJackson2JsonRedisSerializer;
import org.springframework.data.redis.serializer.RedisSerializationContext;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;

import java.time.Duration;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * redis 配置
 *
 * @author admin
 */
@Configuration
public class RedisConfiguration implements ApplicationListener<ApplicationEvent> {
    @Autowired
    private RedisProperties redisProperties;
    /**
     * 100个线程, 无界队列, 超过异常
     */
    private final static ThreadPoolExecutor THREAD_POOL_EXECUTOR = new ThreadPoolExecutor(100, 100, 120, TimeUnit.SECONDS, new LinkedBlockingQueue<>(), new ThreadPoolExecutor.AbortPolicy());

    /**
     * lettuce对象连接池
     *
     * @return
     */
    @Bean
    public LettuceConnectionFactory lettuceConnectionFactory() {
        RedisStandaloneConfiguration configuration = new RedisStandaloneConfiguration();
        configuration.setHostName(redisProperties.getHost());
        configuration.setDatabase(redisProperties.getDatabase());
        configuration.setPort(redisProperties.getPort());
        configuration.setPassword(redisProperties.getPassword());
        //连接池配置
        GenericObjectPoolConfig poolConfig = new GenericObjectPoolConfig();
        RedisProperties.Pool pool = redisProperties.getLettuce().getPool();
        poolConfig.setMaxIdle(20);
        poolConfig.setMinIdle(1);
        poolConfig.setMaxTotal(100);
        poolConfig.setMaxWaitMillis(3000);
        poolConfig.setTimeBetweenEvictionRunsMillis(20000);
        poolConfig.setTestWhileIdle(true);
        LettucePoolingClientConfiguration clientConfiguration = LettucePoolingClientConfiguration.builder()
                .commandTimeout(Duration.ofSeconds(10))
                .poolConfig(poolConfig).build();
        return new LettuceConnectionFactory(configuration, clientConfiguration);
    }

    @Bean
    public RedisTemplate getTemplate(@Qualifier(value = "redisTemplate") RedisTemplate template) {
        template.setConnectionFactory(lettuceConnectionFactory());
        template.setDefaultSerializer(new StringRedisSerializer());
        template.setKeySerializer(new StringRedisSerializer());
        template.setValueSerializer(new StringRedisSerializer());
        template.setHashKeySerializer(new GenericJackson2JsonRedisSerializer());
        template.setHashValueSerializer(new GenericJackson2JsonRedisSerializer());
        return template;
    }

    /**
     * 缓存管理器
     * 1.默认缓存为1天
     *
     * @return
     */
    @Bean
    @Primary
    public RedisCacheManager cacheManager() {
        RedisCacheWriter redisCacheWriter = RedisCacheWriter.nonLockingRedisCacheWriter(lettuceConnectionFactory());
        GenericJackson2JsonRedisSerializer genericJackson2JsonRedisSerializer = new GenericJackson2JsonRedisSerializer();
        RedisCacheConfiguration redisCacheConfiguration = RedisCacheConfiguration.defaultCacheConfig()
                .serializeKeysWith(RedisSerializationContext.SerializationPair.fromSerializer(RedisSerializer.string()))
                .serializeValuesWith(RedisSerializationContext.SerializationPair.fromSerializer(genericJackson2JsonRedisSerializer))
                .entryTtl(Duration.ofDays(1));
        RedisCacheManager redisCacheManager = RedisCacheManager.builder(redisCacheWriter)
                .cacheDefaults(redisCacheConfiguration)
                .withInitialCacheConfigurations(getRedisCacheConfigurationMap()).build();
        return redisCacheManager;
    }

    /**
     * 为指定 key 设置过期时间
     *
     * @return
     */
    private Map<String, RedisCacheConfiguration> getRedisCacheConfigurationMap() {
        Map<String, RedisCacheConfiguration> redisCacheConfigurationMap = new HashMap<>();
        redisCacheConfigurationMap.put(RedisKeys.WECHAT_ACCESS_TOKEN, getRedisCacheConfigurationWithTtl(600));
        // 直播间统计ID集合 默认保存七天
        return redisCacheConfigurationMap;
    }

    private RedisCacheConfiguration getRedisCacheConfigurationWithTtl(Integer seconds) {
        GenericJackson2JsonRedisSerializer genericJackson2JsonRedisSerializer = new GenericJackson2JsonRedisSerializer();
        return RedisCacheConfiguration.defaultCacheConfig()
                .serializeKeysWith(RedisSerializationContext.SerializationPair.fromSerializer(RedisSerializer.string()))
                .serializeValuesWith(RedisSerializationContext.SerializationPair.fromSerializer(genericJackson2JsonRedisSerializer))
                .computePrefixWith(name -> name + ":").entryTtl(Duration.ofSeconds(seconds));
    }


    /**
     * 流监听容器, 这里只是创建, 需要在下面启动
     *
     * @param redisConnectionFactory
     * @return
     */
    @Bean
    public StreamMessageListenerContainer<String, MapRecord<String, String, String>> streamMessageListenerContainer(RedisConnectionFactory redisConnectionFactory) {
        // 创建配置对象
        StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, MapRecord<String, String, String>> listenerContainerOptions = StreamMessageListenerContainer.StreamMessageListenerContainerOptions
                .builder()
                // 一次性最多拉取多少条消息
                .batchSize(100)
                .executor(THREAD_POOL_EXECUTOR)
                .build();
        // 根据配置对象创建监听容器

//        StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer = StreamMessageListenerContainer.create(redisConnectionFactory, listenerContainerOptions);
        StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer = new WjDefaultStreamMessageListenerContainer<>(redisConnectionFactory, listenerContainerOptions);
        return listenerContainer;
    }

    /**
     * 1.spring启动后启动redis监听容器
     * 2.spring结束后关闭redis监听容器
     *
     * @param applicationEvent
     */
    @Override
    public void onApplicationEvent(ApplicationEvent applicationEvent) {
        if (applicationEvent instanceof ApplicationStartedEvent) {
            // 启动redis stream 监听
            ((ApplicationStartedEvent) applicationEvent).getApplicationContext()
                    .getBeanProvider(StreamMessageListenerContainer.class)
                    .ifAvailable(Lifecycle::start);
        }
        if (applicationEvent instanceof ContextClosedEvent) {
            // 启动redis stream 监听
            ((ContextClosedEvent) applicationEvent).getApplicationContext()
                    .getBeanProvider(StreamMessageListenerContainer.class)
                    .ifAvailable(Lifecycle::stop);
        }
    }
}
